package com.zhentao.listener;

import com.alibaba.cloud.commons.lang.StringUtils;
import com.alibaba.fastjson.JSON;
import com.zhentao.common.user.ChargeUserParams;
import com.zhentao.model.pojo.user.history.ChargePointsHistoryChange;
import com.zhentao.model.pojo.user.history.dto.UserHistoryDto;
import com.zhentao.service.ChargePointsHistoryChangeService;
import com.zhentao.utils.RedisIdWorker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Map;

import static com.zhentao.common.user.UserCommon.CHARGE_POINTS_HISTORY_CHANGE_ADD;

/**
 * @ClassName: KafkaUserPointsListener
 * @Author: 振涛教育_Q
 * @Date: 2024/12/23 10:40
 */
@Component
@Slf4j
public class KafkaUserPointsListener {
    @Autowired
    private RedisIdWorker redisIdWorker;
    @Autowired
    private ChargePointsHistoryChangeService historyChangeService;
    @KafkaListener(topics = "charge-user-points")
    public void onMessage(String message, Acknowledgment acknowledgment) {
        log.info("KafkaUserPointsListener监听到消息：{}", message);
        if(StringUtils.isBlank(message)){
            log.error("获取的messages消息为空------------------kafka");
        }
        try {
            // 监听的消息不为空，直接给积分历史表中添加信息
            // 解析msg
            UserHistoryDto historyDto = JSON.parseObject(message, UserHistoryDto.class);
            ChargePointsHistoryChange chargePointsHistoryChange = new ChargePointsHistoryChange();
            // 生成唯一id
            long nextId = redisIdWorker.nextId("charge_points_history_change");
            chargePointsHistoryChange.setId(nextId);
            // 用来判断该积分的信息状态
            Map<Integer, String> statusMap = ChargeUserParams.CHARGE_USER_STATUS_MAP;
            statusMap.keySet().forEach(key -> {
                if(key.equals(historyDto.getStatus())){
                    chargePointsHistoryChange.setChangeInfo(statusMap.get(key));
                }
            });
            chargePointsHistoryChange.setPoints(CHARGE_POINTS_HISTORY_CHANGE_ADD+historyDto.getPoints());
            chargePointsHistoryChange.setCreateTime(new Date());
            chargePointsHistoryChange.setUserId(historyDto.getUserId());
            historyChangeService.save(chargePointsHistoryChange);
            // 手动提交 偏移量
            acknowledgment.acknowledge();
            log.info("一切都已经操作完毕");
        } catch (Exception e) {
            log.info("新增积分变化史.....过程中出现了问题");
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
}
